package com.we.risk.sa;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

public class SaEventLabelStageFirstSampleCmpTest {
    public static final String RELEASEPROP =
            "risk/saevent/stagefirst/saevent_kfk_sample_compare_label_2301220221.properties";
    //    public static final String RELEASEPROP = "risk/saevent_kfk_self_test_prod.properties";
    public static final String PROPSETS = "risk/saevent/stagefirst/saevent_stage1_label.properties";
    public static final String KEYBY = "user_key";
    public static final String BEVTIME = "max_bev_time";
    public static final String RECVTIME = "max_receive_time";
    public static final String LABELK = "label_name";
    public static final String LABELV = "label_value";
    public static final String LABELT = "label_output_time";

    public static Logger LOG = LoggerFactory.getLogger(SaEventLabelHi.class);

    public static final String EVTIDFV = "IdentificationView";
    public static final String EVTCPTC = "ChoiceProtocol";
    public static final String EVTITMC = "ItemChoice";
    public static final String EVTITMINP = "ItemInput";
    public static final String EVTBKBT = "BackButton";

    public static final String EVTAPSTT = "$AppStart";
    public static final String EVTAPVSC = "$AppViewScreen";
    public static final String EVTAPCLI = "$AppClick";
    public static final String EVTAPEND = "$AppEnd";
    /** PageTitile */
    public static final String PGTTBDCAD = "绑定银行卡页";

    public static final String PGTTADCAD = "添加银行卡页";
    public static final String PGTTFCIDT = "人脸识别页";
    public static final String PGTTIDPH = "身份证拍照页";
    public static final String PGTTTCID = "腾讯身份认证页";
    public static final String PGTTIDPG = "身份证页";
    public static final String PGTTEMCT = "紧急联系人页";

    /** LocationPage */
    public static final String LCPGBDCAD = "绑定银行卡";

    public static final String LCPGADCAD = "添加银行卡";
    public static final String LCPGFCIDT = "人脸识别页";
    public static final String LCPGBSIF = "基本信息页";
    public static final String LCPGEMCT = "紧急联系人页";

    /** Item */
    public static final String ITEMIDST = "从事行业";

    public static final String ITEMJOB = "工作岗位";
    public static final String ITEMMRG = "婚姻状况";
    public static final String ITEMUSG = "借款用途";
    public static final String ITEMEDU = "最高学历";
    public static final String ITEMICOM = "月收入";

    public static final String DATEFORMAT = "yyyy-MM-dd";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            ArrayList<String> topicLists = new ArrayList<>();
            topicLists.add(srcKafkaTopic);

            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setTopics(topicLists)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            DataStreamSource<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "Sa_event");

            SingleOutputStreamOperator<String> out =
                    input.uid("SaEvent-source")
                            .filter(new SaEventFilterFunc())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            String user_key = inputJson.getString("distinct_id");
                                            return user_key;
                                        }
                                    })
                            .flatMap(new SaEventFlatMapFunc());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(SaEventLabelStageFirstSampleCmpTest.class.toString());

        } catch (Exception e) {
            //            System.out.println(e.getMessage());
            LOG.error(e.getMessage());
            e.printStackTrace();
        }
    }

    private static class SaEventFilterFunc implements FilterFunction<String> {
        @Override
        public boolean filter(String value) throws Exception {
            boolean res =
                    value.contains(EVTIDFV) || value.contains(EVTCPTC) || value.contains(EVTBKBT);
            res =
                    (res
                            || value.contains(EVTITMC)
                            || value.contains(EVTITMINP)
                            || value.contains(EVTAPVSC));
            res =
                    (res
                            || value.contains(EVTAPCLI)
                            || value.contains(EVTAPSTT)
                            || value.contains(EVTAPEND));

            JSONObject inputJson = JSONObject.parseObject(value);
            Long recv_time = inputJson.getLong("recv_time");
            long now = 1676995200000L; // 2023-02-22 00:00:00
            long startTimeOfDay = getStartTimeOfDay(now);
            long early_time = timeBeforeDays(startTimeOfDay, 30);
            /** 只要当前之间前30天内的数据 按recv_time 过滤 */
            if (recv_time == null || recv_time.equals("") || recv_time < early_time) {
                res = false;
            }
            //            System.out.println("#filter res:" + res + " recv_time: " + recv_time + "
            // now: " + now + " early_time:" + early_time);
            return res;
        }
    }

    private static class SaEventFlatMapFunc extends RichFlatMapFunction<String, String> {
        SimpleDateFormat sdf;
        /** 只存近30天内的数据 */
        MapState<Long, String> resMState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            sdf = new SimpleDateFormat(DATEFORMAT);
            resMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Long, String>(
                                            "resMState", Long.class, String.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                JSONObject inputJson = JSONObject.parseObject(value);

                Long recvTime = inputJson.getLong("recv_time");
                String userKey = inputJson.getString("distinct_id");
                JSONObject properties = inputJson.getJSONObject("properties");
                String app_version = properties.getString("$app_version");

                //            System.out.println("input recvTime: " + (recvTime == null ? "null" :
                // recvTime) + " userkey:" + userKey +
                //                    " appVersion: " + app_version + " event:" +
                // inputJson.getString("event") +
                //                    " bevtime: " + inputJson.getLongValue("time"));
                //            System.out.println("input PageTitle: " +
                // properties.getString("PageTitle") +
                //                    " LocationPage: " + properties.getString("LocationPage"));
                /** 脏数据过滤 */
                if (recvTime != null
                        && !recvTime.equals("")
                        && (app_version != null
                                && (app_version.equals("NULL") || app_version.length() == 5))) {

                    String event = inputJson.getString("event");
                    Long bevTime = inputJson.getLong("time");
                    //                System.out.println("## recvTime: " + recvTime + " bevTime: " +
                    // bevTime);
                    String pageTitle = properties.getString("PageTitle");
                    String locationPage = properties.getString("LocationPage");

                    long mapKey = getStartTimeOfDay(bevTime);
                    long earliestTime = timeBeforeDays(recvTime, 30);
                    //                System.out.println("earliestTime: " + earliestTime + " res: "
                    // + (bevTime >= earliestTime));
                    /** keep 30days data in state */
                    cleanState(resMState, earliestTime);

                    String res = resMState.get(mapKey);

                    JSONObject stateJson = null;
                    if (res != null) {
                        stateJson = JSONObject.parseObject(res);
                    } else {
                        stateJson = initJsonObjState(PROPSETS);
                    }
                    JSONObject resJson = (JSONObject) stateJson.clone();
                    //                System.out.println("event: " + event);
                    switch (event) {
                        case EVTIDFV:
                            idfcationViewEventProcess(bevTime, pageTitle, resJson);
                            break;
                        case EVTBKBT:
                            if (locationPage.equals(LCPGFCIDT)) {
                                // step3
                                setStepPageRange(bevTime, resJson, "step3");
                            }
                            break;
                        case EVTCPTC:
                            if (locationPage.equals(LCPGADCAD) || locationPage.equals(LCPGBDCAD)) {
                                setStepPageRange(bevTime, resJson, "step5");
                            }
                            break;
                        case EVTITMINP:
                            if (locationPage.equals(LCPGEMCT)) {
                                setStepPageRange(bevTime, resJson, "step9");
                            }
                            break;
                        case EVTITMC:
                            itemChoiceEventProcess(inputJson, resJson);
                            break;
                        case EVTAPSTT: // 1 6 6
                        case EVTAPVSC:
                        case EVTAPCLI:
                        case EVTAPEND: // 1 6 6
                            // Last OS
                            resJson.put("ma_varnum_os_event", 1);
                            String os = properties.getString("$os");
                            if (os != null) {
                                int os_code = getOSCode(os);
                                long ma_varnum_latest_os_time =
                                        resJson.getLongValue("ma_varnum_latest_os_time");
                                if (ma_varnum_latest_os_time < bevTime) {
                                    resJson.put("ma_varnum_latest_os_time", bevTime);
                                    resJson.put("ma_varnum_latest_os_code", os_code);
                                }
                            }
                            break;
                        default:
                            //                        LOG.warn("Error Event: " + event + " not
                            // support now!!");
                            break;
                    }

                    long max_bev_time = resJson.getLongValue(BEVTIME);
                    long max_receive_time = resJson.getLongValue(RECVTIME);
                    max_bev_time = (max_bev_time < bevTime ? bevTime : max_bev_time);
                    max_receive_time = (max_receive_time < recvTime ? recvTime : max_receive_time);
                    long label_output_time = System.currentTimeMillis();

                    resJson.put(KEYBY, userKey);
                    resJson.put(BEVTIME, max_bev_time);
                    resJson.put(RECVTIME, max_receive_time);
                    resJson.put(LABELT, label_output_time);

                    JSONObject outJson = new JSONObject();
                    outJson.put(KEYBY, userKey);
                    outJson.put(BEVTIME, max_bev_time);
                    outJson.put(RECVTIME, max_receive_time);
                    outJson.put(LABELT, label_output_time);
                    HashMap<String, Object> tmpLabel = new HashMap<>();

                    Iterator<String> iterator = resJson.keySet().iterator();
                    while (iterator.hasNext()) {
                        String key = iterator.next();
                        Object res_vae = resJson.get(key);
                        Object state_vae = stateJson.get(key);
                        if (!key.equals(KEYBY)
                                && !key.equals(BEVTIME)
                                && !key.equals(RECVTIME)
                                && !key.equals(LABELT)) {
                            if (state_vae != res_vae) {
                                tmpLabel.put(key, res_vae);
                            }
                        }
                    }
                    if (tmpLabel.size() > 0) {
                        Iterator<String> keyIterator = tmpLabel.keySet().iterator();
                        while (keyIterator.hasNext()) {
                            String next = keyIterator.next();
                            //                        System.out.println("##diff key: " + next);
                            outJson.put(LABELV, tmpLabel.get(next));
                            outJson.put(LABELK, next);
                            //                        System.out.println("##outJson user_key: " +
                            // userKey +" outJson size: " + outJson.size() +" outJsonData: " +
                            // outJson.toString());
                            out.collect(outJson.toString(SerializerFeature.WriteMapNullValue));
                        }
                    } else {
                        //                    System.out.println("##outJson user_key: " + userKey +"
                        // outJson size: " + outJson.size() +" outJsonData: " + outJson.toString());
                        out.collect(outJson.toString(SerializerFeature.WriteMapNullValue));
                    }

                    //                System.out.println("##user_key: " + userKey +" ResJson size: "
                    // + resJson.size() +"JsonData: " + resJson.toString());
                    resMState.put(mapKey, resJson.toString(SerializerFeature.WriteMapNullValue));
                } else {
                    //                System.out.println("Warn: filter data!!");
                    LOG.warn("Warn: recv_time or app_version filtered!! ");
                }
            } catch (Exception e) {
                LOG.error("FlatMap Error:" + e.getMessage());
                LOG.error("FlatMap Error input：" + value);
                e.printStackTrace();
            }
        }

        private int getOSCode(String os) {
            int os_code = 0;
            os = os.toUpperCase();
            switch (os) {
                case "ANDROID":
                    os_code = 1;
                    break;
                case "IOS":
                    os_code = 2;
                    break;
                case "HARMONYOS":
                    os_code = 3;
                    break;
                default:
                    os_code = 0;
                    break;
            }
            return os_code;
        }

        private void itemChoiceEventProcess(JSONObject inputJson, JSONObject resJson) {
            String event = inputJson.getString("event");
            //            long bevTime = inputJson.getLongValue("time") / 1000 * 1000;
            long bevTime = inputJson.getLongValue("time");
            JSONObject properties = inputJson.getJSONObject("properties");
            String locationPage = properties.getString("LocationPage");

            if (locationPage.equals(LCPGBSIF)) {
                String item = properties.getString("Item");
                String operationContent = properties.getString("OperationContent");
                //                System.out.println(" item: "+ item + " OperationContent: " +
                // operationContent) ;
                if (item != null) {
                    switch (item) {
                            /** 需要考虑填空值的情况 */
                        case ITEMEDU: // 学历
                            resJson.put(SaEventCodeProcess.getEduCode(operationContent), 1);
                            resJson.put("ma_varnum_education_code00", 1);
                            break;
                        case ITEMIDST: // 行业
                            resJson.put(SaEventCodeProcess.getIdstryCode(operationContent), 1);
                            resJson.put("ma_varnum_has_industry_code00", 1);
                            break;
                        case ITEMJOB: // 工作
                            resJson.put(SaEventCodeProcess.getJobCode(operationContent), 1);
                            resJson.put("ma_varnum_job_code00", 1);
                            break;
                        case ITEMMRG: // 婚姻
                            resJson.put(SaEventCodeProcess.getMrgCode(operationContent), 1);
                            resJson.put("ma_varnum_marridge_code00", 1);
                            break;
                        case ITEMUSG: // 用途
                            resJson.put(SaEventCodeProcess.getUsageCode(operationContent), 1);
                            resJson.put("ma_varnum_usage_code00", 1);
                            break;
                        case ITEMICOM: // 收入均值 + 最后一次填写的收入值 不区分时间戳，
                            Integer income = SaEventCodeProcess.getIncomeFix2Code(operationContent);
                            if (income != null) {
                                // 新特征: ma_varnum_income_avg_fix1,ma_varnum_income_final_fix2
                                int ma_varnum_income_sum_fix2 =
                                        resJson.getIntValue("ma_varnum_income_sum_fix2");
                                int ma_varnum_income_cnt_fix2 =
                                        resJson.getIntValue("ma_varnum_income_cnt_fix2");
                                ma_varnum_income_sum_fix2 += income;
                                ma_varnum_income_cnt_fix2 += 1;

                                resJson.put("ma_varnum_income_sum_fix2", ma_varnum_income_sum_fix2);
                                resJson.put("ma_varnum_income_cnt_fix2", ma_varnum_income_cnt_fix2);

                                long ma_varnum_income_final_fix2_time =
                                        resJson.getLongValue("ma_varnum_income_final_time_fix2");
                                if (bevTime > ma_varnum_income_final_fix2_time) {
                                    resJson.put("ma_varnum_income_final_time_fix2", bevTime);
                                    resJson.put("ma_varnum_income_final_fix2", income);
                                }
                            }
                            break;
                        default:
                            //                                    System.out.println("Item Error: "
                            // + event + " " + locationPage + " " + item + " not support now!!");
                            //                            LOG.warn("Item Error: " + event + " " +
                            // locationPage + " " + item + " not support now!!");
                            break;
                    }
                }
            }
        }

        private void idfcationViewEventProcess(Long bevTime, String pageTitle, JSONObject resJson) {
            switch (pageTitle) {
                case PGTTIDPH: // step1
                case PGTTTCID:
                case PGTTIDPG:
                    setStepPageRange(bevTime, resJson, "step1");
                    break;
                case PGTTFCIDT: // step3
                    setStepPageRange(bevTime, resJson, "step3");
                    break;
                case PGTTADCAD: // step5
                case PGTTBDCAD:
                    setStepPageRange(bevTime, resJson, "step5");
                    break;
                case PGTTEMCT: // step9
                    setStepPageRange(bevTime, resJson, "step9");
                    break;
                default:
                    //                    LOG.warn("Event:" + EVTIDFV + " not support pageTitle: " +
                    // pageTitle + " now!!");
                    break;
            }
        }

        private void setStepPageRange(Long bevTime, JSONObject resJson, String step) {
            String minKey = "ma_varnum_" + step + "_min_time";
            String maxKey = "ma_varnum_" + step + "_max_time";

            long minTime = resJson.getLongValue(minKey);
            long maxTime = resJson.getLongValue(maxKey);

            if (minTime == 0 || bevTime < minTime) {
                resJson.put(minKey, bevTime);
            }
            if (bevTime > maxTime) {
                resJson.put(maxKey, bevTime);
            }
        }

        private void cleanState(MapState<Long, String> state, long earliestTime) throws Exception {
            if (!state.isEmpty()) {
                ArrayList<Long> timeLists = new ArrayList<>();
                Iterator<Long> iterator = state.keys().iterator();
                while (iterator.hasNext()) {
                    Long timeKey = iterator.next();
                    if (timeKey < earliestTime) {
                        timeLists.add(timeKey);
                    }
                }
                int listSize = timeLists.size();
                for (int i = 0; i < listSize; i++) {
                    state.remove(timeLists.get(i));
                }
            }
        }

        private static JSONObject initJsonObjState(String propFile) throws IOException {
            JSONObject res = new JSONObject();
            InputStream resourceAsStream =
                    SaEventLabelStageFirstSampleCmpTest.class
                            .getClassLoader()
                            .getResourceAsStream(propFile);
            ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
            Set<Object> propSets = reader.getProperties().keySet();

            for (Object key : propSets) {
                String typeValue = reader.get((String) key);
                String[] splits = typeValue.split(":");
                if (splits.length == 2) {
                    String type = splits[0];
                    String value = splits[1];
                    jsonObjectSetValue(res, type, (String) key, value);
                } else {
                    LOG.error("split Error: " + splits.length);
                }
            }
            return res;
        }

        private static void jsonObjectSetValue(
                JSONObject jsonObj, String type, String key, String value) {
            if (!value.equals("null")) {
                switch (type) {
                    case "Byte":
                        jsonObj.put(key, Byte.parseByte(value));
                        break;
                    case "Long":
                        jsonObj.put(key, Long.parseLong(value));
                        break;
                    case "String":
                        jsonObj.put(key, value);
                        break;
                    case "Float":
                        jsonObj.put(key, Float.parseFloat(value));
                        break;
                    case "Integer":
                        jsonObj.put(key, Integer.parseInt(value));
                        break;
                    default:
                        LOG.error("Not support this type: " + type);
                        break;
                }
            } else {
                jsonObj.put(key, null);
            }
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(KEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }

    private static long timeBeforeDays(long now, int days) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(now);
        calendar.add(Calendar.DATE, -days);
        long res = calendar.getTimeInMillis();
        return res;
    }

    private static long getStartTimeOfDay(long now) {
        SimpleDateFormat sdf = new SimpleDateFormat(DATEFORMAT);
        String date = sdf.format(new Date(now));
        long res = 0;
        try {
            res = sdf.parse(date).getTime();
        } catch (Exception e) {
            LOG.error("sdf parse date error! time: " + now);
            e.printStackTrace();
        }
        return res;
    }
}
